package com.streamsets.datacollector.execution.mq;

import com.google.gson.Gson;
import com.streamsets.datacollector.event.json.MetricRegistryJson;
import com.streamsets.datacollector.execution.mq.config.MqConfigUtil;
import com.streamsets.datacollector.metrics.MetricsEventListener;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MqStateMetricsNotifier implements MetricsEventListener {

  private static final Logger LOG = LoggerFactory.getLogger(MqStateMetricsNotifier.class);

  private String baseConfDir;

  public MqStateMetricsNotifier(String baseConfDir) {
    this.baseConfDir = baseConfDir;
  }

  public MqStateMetricsNotifier() {
  }


  @Override
  public void notification(String metrics) {

    MqConfigUtil mqConfigUtil = new MqConfigUtil(baseConfDir);
    try{
      Gson gson = new Gson();
      MetricRegistryJson metricRegistryJson = gson.fromJson(metrics, MetricRegistryJson.class);
      if(metricRegistryJson != null && metricRegistryJson.getCounters() != null){
        KafkaProducer<String, String> producer = mqConfigUtil.getKafkaProducer();
        ProducerRecord<String, String> record = new ProducerRecord<>(mqConfigUtil.getMetricsEventTopicName()
          , new Gson().toJson(metricRegistryJson.getCounters()));
        producer.send(record);
      }
    }catch (Exception e){
      LOG.error(e.getMessage(), e);
    }
  }




}
